package com.ruoyi.system.service.impl;

import java.util.Date;
import java.util.List;
import java.util.Properties;

import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.common.utils.DateUtils;
import com.ruoyi.system.domain.DataSource;
import com.ruoyi.system.domain.TaskExecutionRecord;
import com.ruoyi.system.mapper.DataSourceMapper;
import com.ruoyi.system.service.ITaskExecutionRecordService;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.ruoyi.system.mapper.CollectionTaskMapper;
import com.ruoyi.system.domain.CollectionTask;
import com.ruoyi.system.service.ICollectionTaskService;
import com.ruoyi.common.core.text.Convert;



/**
 * 采集任务信息Service业务层处理
 *
 * @author ruoyi
 * @date 2025-04-30
 */
@Service
public class CollectionTaskServiceImpl implements ICollectionTaskService
{
    @Autowired
    private CollectionTaskMapper collectionTaskMapper;
    @Autowired
    private DataSourceMapper dataSourceMapper;
    @Autowired
    private ITaskExecutionRecordService iTaskExecutionRecordService;

    /**
     * 查询采集任务信息
     *
     * @param id 采集任务信息主键
     * @return 采集任务信息
     */
    @Override
    public CollectionTask selectCollectionTaskById(Long id)
    {
        return collectionTaskMapper.selectCollectionTaskById(id);
    }

    /**
     * 查询采集任务信息列表
     *
     * @param collectionTask 采集任务信息
     * @return 采集任务信息
     */
    @Override
    public List<CollectionTask> selectCollectionTaskList(CollectionTask collectionTask)
    {
        return collectionTaskMapper.selectCollectionTaskList(collectionTask);
    }

    /**
     * 新增采集任务信息
     *
     * @param collectionTask 采集任务信息
     * @return 结果
     */
    @Override
    public int insertCollectionTask(CollectionTask collectionTask)
    {
        collectionTask.setCreateTime(DateUtils.getNowDate());
        return collectionTaskMapper.insertCollectionTask(collectionTask);
    }

    /**
     * 修改采集任务信息
     *
     * @param collectionTask 采集任务信息
     * @return 结果
     */
    @Override
    public int updateCollectionTask(CollectionTask collectionTask)
    {
        collectionTask.setUpdateTime(DateUtils.getNowDate());
        return collectionTaskMapper.updateCollectionTask(collectionTask);
    }

    /**
     * 批量删除采集任务信息
     *
     * @param ids 需要删除的采集任务信息主键
     * @return 结果
     */
    @Override
    public int deleteCollectionTaskByIds(String ids)
    {
        return collectionTaskMapper.deleteCollectionTaskByIds(Convert.toStrArray(ids));
    }

    /**
     * 删除采集任务信息信息
     *
     * @param id 采集任务信息主键
     * @return 结果
     */
    @Override
    public int deleteCollectionTaskById(Long id)
    {
        return collectionTaskMapper.deleteCollectionTaskById(id);
    }

    @Override
    public int executeTask(Long id) {
        //查询任务，获取任务信息
        CollectionTask task = collectionTaskMapper.selectCollectionTaskById(id);
//        if("1".equals(task.getTaskStatus())) {
//            throw new ServiceException("任务正在执行，请稍后再试！");
//        }
        //todo 采数逻辑
//        try {
            // 创建 SparkSession 对象
//            SparkSession spark = SparkSession.builder()
//                    .appName("dataPlateForm")
//                    .master("local[*]")
//                    .config("spark.sql.codegen", "false")
//                    .getOrCreate();
//
//            //获取数据源信息
//            DataSource dataSourceVo = new DataSource();
//            dataSourceVo.setJdbcUrl(task.getDataSource());
//            List<DataSource> dataSources = dataSourceMapper.selectDataSourceList(dataSourceVo);
//            if(dataSources == null || dataSources.size() == 0) {
//                throw new ServiceException("数据源为空，请重新设置数据源！");
//            }
//            String url = task.getDataSource();
//            DataSource dataSource = dataSources.get(0);
//
//            Properties properties = new Properties();
//            properties.setProperty("user", dataSource.getUser());
//            properties.setProperty("password", dataSource.getPassword());
//            properties.setProperty("driver", "com.mysql.cj.jdbc.Driver");
//
//            // 执行 SQL 查询
//            String sql = task.getExecutionSql();
//            Dataset<Row> resultDF = spark.read()
//                    .jdbc(url, "(" + sql + ") AS subquery", properties);
//
//            // 将结果插入到 目标表中 表
//            resultDF.write()
//                    .mode("append")
//                    .jdbc(url, task.getTargetTable(), properties);
//
//            // 停止 SparkSession
//            spark.stop();
        //更新任务状态
        CollectionTask collectionTask = new CollectionTask();
        collectionTask.setId(id);
        collectionTask.setTaskStatus("1");
        updateCollectionTask(collectionTask);

        //增加执行记录
        TaskExecutionRecord TaskExecutionRecord = new TaskExecutionRecord();
        TaskExecutionRecord.setTaskId(id);

        //执行成功
        TaskExecutionRecord.setExecutionStatus("2");
        TaskExecutionRecord.setErrorMessage("执行成功");
        TaskExecutionRecord.setExecutionTime(new Date());
        iTaskExecutionRecordService.insertTaskExecutionRecord(TaskExecutionRecord);
        return 1;
    }
}
